package com.meihua.code.leader;

import com.meihua.code.leader.domain.NodeRoleEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Ratf协议
 * 可能存在的问题
 * 如果所有节点同时发起选举 则可能导致 每个节点投给自己 导致选不出leader 所以选举阶段需避免 同时投票这种情况出现
 *
 * 领导者  提出提议 日志复制 定时发送心跳
 * 跟随着
 * 候选者
 * 设定 每台节点 设置随机的超时时间 避免节点同时成为候选人 初始化时也需要设置
 * 如果有多个candidate，最终一定会有一个被选举为leader
 * 如果多个candidate同时发起了选举，导致都没有获得大多数选票时，每一个candidate会随机等待一段时间后重新发起新一轮投票（一般是随机等待150-300ms）
 *
 *
 *
 * @author meihua
 * @version 1.0
 * @date 2021/4/26 20:28
 */
public class RatfTest {

    private static final ThreadPoolExecutor pool=new ThreadPoolExecutor(3,3,0,TimeUnit.MINUTES,new ArrayBlockingQueue<Runnable>(100));

    public static void main(String[] args) {
        List<Raft> nodes=new ArrayList<Raft>(3);
        nodes.add(new Raft("A节点", NodeRoleEnum.candidate));
        nodes.add(new Raft("B节点",NodeRoleEnum.follower));
        nodes.add(new Raft("C节点",NodeRoleEnum.follower));
        for (Raft node :nodes){
            node.setNodes(nodes);
            pool.execute(node);
        }
    }

}

class Raft implements Runnable{

    private final static int POLL_TIME=10;
    private final  static  Logger logger= LoggerFactory.getLogger(Raft.class);
    /**
     * 节点角色
     **/
    private NodeRoleEnum role;

    private final String nodeName;
    /**
     * 节点触发选举的休眠时间
     **/
    private int sleepTime;
    /**
     * 心跳超时时间
     **/
    private long beatTimeOut;

    /**
     * 上次心跳时间
     **/
    private long lastBeatTime;
    /**
     * 任期 选举轮数 日志复制比对时使用
     **/
    private AtomicInteger term=new AtomicInteger(0);

    /**
     * leader 节点的信息
     **/
    private Raft leader;

    /**
     * 自己票数
     **/
    private AtomicInteger localPollCount = new AtomicInteger(1);

    /**
     * 总投票数
     **/
    private AtomicInteger sumPollCount=new AtomicInteger(0);

    /**
     * 其他节点 模拟交互通信
     **/
    private List<Raft> nodes;

    /**
     * 日志复制队列
     **/
    private LinkedBlockingDeque queue=new LinkedBlockingDeque<Object>();

    @Override
    public void run() {
        try {
            while (true){

            //检查是否超时
            if (System.currentTimeMillis()-lastBeatTime>beatTimeOut){
                logger.info("leader {} timeout 触发选举 time {}",leader.nodeName,System.currentTimeMillis());
                election();
            }

//            //不是leader 且leader 存在 心跳检查
//            if (role!=NodeRoleEnum.Leader&&leader!=null){
//                //leader 超时 触发重新选举
//                if (System.currentTimeMillis()-leaderTime.get()>100){
//                    leaderName=null;
//                    localPollCount.compareAndSet(0,1);
//
//                    break;
//                }
//                logger.info("{} {} ,{} 心跳成功！ leader 是 {}",System.currentTimeMillis()-lastBeatTime,nodeName,lastBeatTime,leaderName);
//                Thread.sleep(pollTime);
//            }

                //是leader
                if (role==NodeRoleEnum.Leader){
                    int succeedCount=0;
                    for (Raft node:nodes){
                        //日志复制与心跳
                        if(!node.notifyEvent(this.nodeName)) {
                            logger.error("心跳失败！");
                            succeedCount++;
                        }
                    }
//                    //成功数不超过总结点一半
                    if (succeedCount<(nodes.size()/2)){
                        logger.info("通知超时");
//                        localPollCount.compareAndSet(0,1);
//                        break;
                    }
//                    Thread.sleep(pollTime*2);
                }
                Thread.sleep(POLL_TIME);
          }
        }catch (Exception e){

        }
    }
    //投票请求
    public boolean votingEvent(String nodeName,int term){
        logger.info("{}：向{}发起了投票请求, term {}",nodeName,this.nodeName,term);
        if (localPollCount.compareAndSet(1,0)){
            logger.info("{} 向 {}投票成功",this.nodeName,nodeName);
            return true;
        }else {
            logger.info("{}你没有多余的票了 向 {}投票失败！",this.nodeName,nodeName);
            return false;
        }
    }

    public boolean notifyLeader(Raft leader){
            this.leader=leader;
            logger.info("{}成为Leader",this.nodeName);
            logger.info("{} 通知{}成为 leader！ ",nodeName,this.nodeName);
            return true;
    }

    public boolean notifyEvent(String nodeName){
        if (this.leader!=null&&this.leader.nodeName.equals(nodeName)){
            logger.info("收到leader 心跳 节点 {} time{}",nodeName,System.currentTimeMillis());
            lastBeatTime=System.currentTimeMillis();
            return true;
        }
       return false;
    }

     /**
      * 选举
      **/
    public void election(){
        //开始选举
        try {
            role=NodeRoleEnum.candidate;
            leader=null;
            Thread.sleep(sleepTime);
            if (leader!=null){
                return;
            }
            logger.info("{}发起投票 {}",this.nodeName,leader);
            this.role=NodeRoleEnum.candidate;
            sumPollCount.set(0);
            term.incrementAndGet();
            for (Raft node:nodes){
                //申请投票成功
                if(node.votingEvent(nodeName, term.get())){
                    sumPollCount.incrementAndGet();
                }
            }
            //超过半数投票成为leader
            if (sumPollCount.get()>(nodes.size()/2)){
                for (Raft node:nodes){
                    if(node.notifyLeader(this)){
                        logger.info("写入成功！");
                    }else {
                        logger.error("通知 {}失败！ Leader: {}",node.nodeName,this.nodeName);
                    }
                }
            }
        }catch (Exception e){

        }

    }


    public Raft(String name,NodeRoleEnum role) {
        this.nodeName=name;
        this.role=role;
        this.sleepTime =(int)(Math.random());
        logger.info("节点 {} 休眠时间{}： ms",name,sleepTime);
    }

    public void setNodes(List<Raft> nodes) {
        this.nodes = nodes;
    }
}
